package com.small.push.consumer.impl;

import com.small.common.manager.redis.RedisClient;
import com.small.push.bo.WeChatTemplateMsg;
import com.small.push.config.MsgPushConfig;
import com.small.push.config.Topic;
//import com.small.push.config.TopicConfig;
import com.small.push.config.WeChatTemplateMsgTopic;
import com.small.push.consumer.MsgConsumer;
import com.small.push.manager.WeChatTemplateMsgManager;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import static com.small.common.constant.GlobalConstants.*;

/**
 * 使用Redis实现的消费者
 * @author dingyou
 * @version 1.0
 * @date 2019-01-10
 */
@Slf4j
@Service
public class RedisMsgConsumer implements MsgConsumer {
    private static final AtomicBoolean run = new AtomicBoolean(true);
    /** 默认消息推送执行线程池大小,等于cpu核数 */
    private static final int CORE_THREAD_SIZE = Runtime.getRuntime().availableProcessors();
    /** 最大消息推送执行线程池大小, 生产环境是4核,这里就是128 */
    private static final int MAX_THREAD_SIZE = CORE_THREAD_SIZE << 5;
    /** 队列深度2048(生产环境4核,2048/4*100/1000=51.2秒,即1分钟内能处理完就不增加线程) */
    private static final int QUEUE_SIZE = 1 << 11;
    /** 执行者采用CallerRunsPolicy,处理线程+队列全满的时候将任务丢给调用线程处理 */
    private static final ThreadPoolExecutor msgPushExecutor = new ThreadPoolExecutor(CORE_THREAD_SIZE, MAX_THREAD_SIZE, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(QUEUE_SIZE), new ThreadPoolExecutor.CallerRunsPolicy());

//    private final TopicConfig topicConfig;
    private final MsgPushConfig pushConfig;
    private final RedisClient redisClient;
    private final WeChatTemplateMsgManager weChatTemplateMsgManager;

    public RedisMsgConsumer( MsgPushConfig pushConfig, RedisClient redisClient,WeChatTemplateMsgManager weChatTemplateMsgManager) {
//        this.topicConfig = topicConfig;
        this.pushConfig = pushConfig;
        this.redisClient = redisClient;
        this.weChatTemplateMsgManager = weChatTemplateMsgManager;
    }

    /**
     * 消费消息
     */
    @Override
    @PostConstruct
    public void consume() {
        if (pushConfig.isOpen()) {
            log.info("启动消息消费线程...");
            // 目前有微信、邮件、短信,分别获取他们的主题,并挨个启动消费线程
//            TopicConfig.ConsumeTopic consume = topicConfig.getConsume();
//            Set<WeChatTemplateMsgTopic> wechatTemplateMsgTopics = consume.getWechatTemplateMsgTopics();
//            if (CollectionUtils.isNotEmpty(wechatTemplateMsgTopics)) {
//                wechatTemplateMsgTopics.forEach(this::processWechat);
//            }
//            Set<Topic> emailMsgTopics = consume.getEmailMsgTopics();
//            if (CollectionUtils.isNotEmpty(emailMsgTopics)) {
//                emailMsgTopics.forEach(this::processEmail);
//            }
//            Set<Topic> smsMsgTopics = consume.getSmsMsgTopics();
//            if (CollectionUtils.isNotEmpty(smsMsgTopics)) {
//                smsMsgTopics.forEach(this::processSms);
//            }
            brpop();
            log.info("项目启动。。。");
        } else {
            log.info("消息推送已关闭...");
        }
    }
    
    public void brpop(){
        new Thread(new Runnable() {
            public void run() {
                log.info("消费消息线程启动。。。");
                while (run.get()) {
                    String result = redisClient.brpop("redis-topic:topic2");    
                    log.info("result------------"+result);
                }
                
            }
        }).start();
    }

    @PreDestroy
    public void destroy() {
        log.info("@PreDestroy#####################");
        run.set(false);
        msgPushExecutor.shutdown();
    }

    /**
     * 处理短信消息
     * @param topic
     */
    private void processSms(Topic topic) {
        int consumeThreads = topic.getConsumeThreads();
        if (consumeThreads <= 0) {
            log.warn("主题 {} 配置处理线程数为 {} , 请确认是否不需要消费该主题...", topic, consumeThreads);
            return;
        }
        // TODO 暂时没有短信消息
    }

    /**
     * 处理电子邮件消息
     * @param topic
     */
    private void processEmail(Topic topic) {
        int consumeThreads = topic.getConsumeThreads();
        if (consumeThreads <= 0) {
            log.warn("主题 {} 配置处理线程数为 {} , 请确认是否不需要消费该主题...", topic, consumeThreads);
            return;
        }
        // TODO 暂时没有邮件消息
    }

    /**
     * 处理微信模板消息
     * @param topic
     */
    private void processWechat(WeChatTemplateMsgTopic topic) {
        int consumeThreads = topic.getConsumeThreads();
        if (consumeThreads <= 0) {
            log.warn("主题 {} 配置处理线程数为 {} , 请确认是否不需要消费该主题...", topic, consumeThreads);
            return;
        }
        // 根据配置的线程数启动消费者线程
        for (int i = 0; i < consumeThreads; i++) {
            String name = topic.getTopic() + " consumer-" + i;
            new Thread(new ProcessWechatMsgRunnable(topic), name).start();
            log.info("消息消费线程 {} 已启动...", name);
        }
    }

    /**
     * 处理微信模板消息
     */
    class ProcessWechatMsgRunnable implements Runnable {
        private final WeChatTemplateMsgTopic topic;
        private final String topicKey;

        ProcessWechatMsgRunnable(WeChatTemplateMsgTopic topic) {
            this.topic = topic;
            this.topicKey = topic.getTopic();
        }

        @Override
        public void run() {
            while (run.get()) {
                WeChatTemplateMsg templateMsg = redisClient.brpop(topicKey, WeChatTemplateMsg.class);
                if (Objects.isNull(templateMsg)) {
                    continue;
                }
                msgPushExecutor.execute(() -> {
                    String url = topic.getUrl();
                    // 如果没有详情url就直接发送,否则就需要检查url参数并添加上url参数
                    if (StringUtils.isBlank(url)) {
                        weChatTemplateMsgManager.pushMsg(templateMsg.getToUser(), topic.getTemplateId(), null,
                                templateMsg.getData(), templateMsg.getFromUser());
                    } else {
                        Map<String, Object> params = templateMsg.getUrlParams();
                        if (MapUtils.isNotEmpty(params)) {
                            StringBuilder urlBuilder = new StringBuilder(url).append(INTERROGATION_MARK);
                            params.forEach((K, V) -> urlBuilder.append(K).append(EQUAL_SIGN).append(V).append(AND_SIGN));
                            url = urlBuilder.deleteCharAt(urlBuilder.lastIndexOf(AND_SIGN)).toString();
                        }
                        weChatTemplateMsgManager.pushMsg(templateMsg.getToUser(), topic.getTemplateId(), url,
                                templateMsg.getData(), templateMsg.getFromUser());
                    }
                });
            }
        }
    }
}
